package dht.network; /**
 * File:   Network.java
 * Author: Gvennet
 */

/** todo write javadoc */

import dht.identifier.identifier.Identifier;
import dht.identifier.identifier.Identifiers;
import dht.node.DHTDataRow;
import dht.node.DHTNodeContacts;
import dht.node.DHTNodeRow;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public final class Network
{

	private DatagramSocket datagramSocket;

	private final List<NetworkListener> listeners;

	private final int localPort;

	private final InetAddress localAddress;

	private final Identifier localId;

	private AtomicInteger messageCounter;

	private final Map<Integer, DHTNodeContacts> outboundMessages;

	private final ScheduledThreadPoolExecutor scheduledExecutor;

	private final long timeout;

	/* статистика */

	private int pingRequestIn = 0;

	private int pingResponseIn = 0;

	private int getNodesRequestIn = 0;

	private int getNodesResponseIn = 0;

	private int putValueRequestIn = 0;

	private int putValueResponseIn = 0;

	private int getValueRequestIn = 0;

	private int getValueResponseIn = 0;

	private int getValueResponseNodesIn = 0;

	private int idLen = 20;

	private int pingRequestOut = 0;

	private int pingResponseOut = 0;

	private int getNodesRequestOut = 0;

	private int getNodesResponseOut = 0;

	private int putValueRequestOut = 0;

	private int putValueResponseOut = 0;

	private int getValueRequestOut = 0;

	private int getValueResponseOut = 0;

	private int getValueResponseNodesOut = 0;

	private int bytesIn = 0;

	private int bytesOut = 0;

	private final LinkedList<int[]> statistics = new LinkedList<int[]>();

	private final Map<Identifier, Integer> findValueStatistics = new HashMap<Identifier, Integer>();

	private boolean receiving = true;

	public Network(InetAddress ia, int port, ScheduledThreadPoolExecutor scheduledExecutor, Identifier identifier, long timeout)
	{
		listeners = new ArrayList<NetworkListener>();

		try
		{
			datagramSocket = new DatagramSocket(port, ia);
		}
		catch (SocketException e)
		{
			throw new RuntimeException(e);
		}

		this.localPort = datagramSocket.getLocalPort();
		this.localAddress = datagramSocket.getLocalAddress();
		if (identifier != null)
		{
			localId = identifier;
		}
		else
		{
			localId = Identifiers.createIdentifier(Arrays.toString(getLocalAddress().getAddress()) + getLocalPort(), idLen);
		}
		messageCounter = new AtomicInteger();
		outboundMessages = new ConcurrentHashMap<Integer, DHTNodeContacts>();
		this.scheduledExecutor = scheduledExecutor;
		this.timeout = timeout;

	}

	public Network(InetAddress ia, int port, ScheduledThreadPoolExecutor scheduledExecutor, Identifier identifier, long timeout, int idLen)
	{
		listeners = new ArrayList<NetworkListener>();

		try
		{
			datagramSocket = new DatagramSocket(port, ia);
		}
		catch (SocketException e)
		{
			throw new RuntimeException(e);
		}

		this.localPort = datagramSocket.getLocalPort();
		this.localAddress = datagramSocket.getLocalAddress();
		if (identifier != null)
		{
			localId = identifier;
		}
		else
		{
			localId = Identifiers.createIdentifier(Arrays.toString(getLocalAddress().getAddress()) + getLocalPort(), idLen);
		}
		messageCounter = new AtomicInteger();
		outboundMessages = new ConcurrentHashMap<Integer, DHTNodeContacts>();
		this.scheduledExecutor = scheduledExecutor;
		this.timeout = timeout;
		this.idLen = idLen;

	}

	public int getPingRequestIn()
	{
		return pingRequestIn;
	}

	public int getPingResponseIn()
	{
		return pingResponseIn;
	}

	public int getGetNodesRequestIn()
	{
		return getNodesRequestIn;
	}

	public int getGetNodesResponseIn()
	{
		return getNodesResponseIn;
	}

	public int getPutValueRequestIn()
	{
		return putValueRequestIn;
	}

	public int getPutValueResponseIn()
	{
		return putValueResponseIn;
	}

	public int getGetValueRequestIn()
	{
		return getValueRequestIn;
	}

	public int getGetValueResponseIn()
	{
		return getValueResponseIn;
	}

	public int getGetValueResponseNodesIn()
	{
		return getValueResponseNodesIn;
	}

	public int getPingRequestOut()
	{
		return pingRequestOut;
	}

	public int getPingResponseOut()
	{
		return pingResponseOut;
	}

	public int getGetNodesRequestOut()
	{
		return getNodesRequestOut;
	}

	public int getGetNodesResponseOut()
	{
		return getNodesResponseOut;
	}

	public int getPutValueRequestOut()
	{
		return putValueRequestOut;
	}

	public int getPutValueResponseOut()
	{
		return putValueResponseOut;
	}

	public int getGetValueRequestOut()
	{
		return getValueRequestOut;
	}

	public int getGetValueResponseOut()
	{
		return getValueResponseOut;
	}

	public int getGetValueResponseNodesOut()
	{
		return getValueResponseNodesOut;
	}

	public int getBytesIn()
	{
		return bytesIn;
	}

	public int getBytesOut()
	{
		return bytesOut;
	}

	public Identifier getLocalId()
	{
		return localId;
	}

	public InetAddress getLocalAddress()
	{
		return localAddress;
	}

	public int getLocalPort()
	{
		return localPort;
	}

	public void addNetworkListener(NetworkListener listener)
	{
		listeners.add(listener);
	}

	public void sendPingRequest(InetAddress remoteAddress, int remotePort) throws IOException
	{
		if (messageCounter.get() == Integer.MAX_VALUE)
		{
			messageCounter.set(0);
		}

		int messageId = messageCounter.incrementAndGet();
		ByteBuffer byteBuffer = encodePingRequest(messageId);

		bytesOut = bytesOut + byteBuffer.limit();

		outboundMessages.put(messageId, new DHTNodeContacts(remoteAddress, remotePort));
		scheduledExecutor.schedule(new TimeoutTask(messageId), timeout, TimeUnit.SECONDS);

		pingRequestOut++;

		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);

		datagramSocket.send(p);
	}

	public void sendPingResponse(InetAddress remoteAddress, int remotePort, int messageId) throws IOException
	{
		ByteBuffer byteBuffer = encodePingResponse(messageId);

		bytesOut = bytesOut + byteBuffer.limit();

		pingResponseOut++;

		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);

		datagramSocket.send(p);

	}

	public void sendGetNodesRequest(InetAddress remoteAddress, int remotePort, Identifier searchingId)
			throws IOException
	{
		if (messageCounter.get() == Integer.MAX_VALUE)
		{
			messageCounter.set(0);
		}

		int messageId = messageCounter.incrementAndGet();
		ByteBuffer byteBuffer = encodeGetNodesRequest(messageId, searchingId);

		bytesOut = bytesOut + byteBuffer.limit();

		outboundMessages.put(messageId, new DHTNodeContacts(remoteAddress, remotePort));
		scheduledExecutor.schedule(new TimeoutTask(messageId), timeout, TimeUnit.SECONDS);

		getNodesRequestOut++;

		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);

		datagramSocket.send(p);
	}

	public void sendGetNodesResponse(InetAddress remoteAddress, int remotePort, int messageId, List<DHTNodeRow> nodes)
			throws IOException
	{

		ByteBuffer byteBuffer = encodeGetNodesResponse(messageId, nodes);

		bytesOut = bytesOut + byteBuffer.limit();

		getNodesResponseOut++;

		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);

		datagramSocket.send(p);
	}

	public void sendPutValueRequest(InetAddress remoteAddress, int remotePort, Identifier valueId, String value)
			throws IOException
	{

		if (messageCounter.get() == Integer.MAX_VALUE)
		{
			messageCounter.set(0);
		}
		int messageId = messageCounter.incrementAndGet();
		ByteBuffer byteBuffer = encodePutValueRequest(messageId, valueId, value);

		bytesOut = bytesOut + byteBuffer.limit();

		outboundMessages.put(messageId, new DHTNodeContacts(remoteAddress, remotePort));
		scheduledExecutor.schedule(new TimeoutTask(messageId), timeout, TimeUnit.SECONDS);

		putValueRequestOut++;

		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);

		datagramSocket.send(p);
	}

	public void sendPutValueResponse(InetAddress remoteAddress, int remotePort, int messageId) throws IOException
	{

		ByteBuffer byteBuffer = encodePutValueResponse(messageId);

		bytesOut = bytesOut + byteBuffer.limit();

		putValueResponseOut++;

		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);

		datagramSocket.send(p);
	}

	public void sendGetValueRequest(InetAddress remoteAddress, int remotePort, Identifier valueId) throws IOException
	{

		if (messageCounter.get() == Integer.MAX_VALUE)
		{
			messageCounter.set(0);
		}
		int messageId = messageCounter.incrementAndGet();

		ByteBuffer byteBuffer = encodeGetValueRequest(messageId, valueId);

		bytesOut = bytesOut + byteBuffer.limit();

		getValueRequestOut++;

		if (findValueStatistics.containsKey(valueId))
		{
			findValueStatistics.put(valueId, findValueStatistics.get(valueId) + 1);
		}
		else
		{
			findValueStatistics.put(valueId, 1);
		}

		outboundMessages.put(messageId, new DHTNodeContacts(remoteAddress, remotePort));
		scheduledExecutor.schedule(new TimeoutTask(messageId), timeout, TimeUnit.SECONDS);

		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);
		datagramSocket.send(p);
	}

	public void sendGetValueResponse(InetAddress remoteAddress, int remotePort, int messageId, List<DHTDataRow> values)
			throws IOException
	{

		ByteBuffer byteBuffer = encodeGetValueResponse(messageId, values);

		bytesOut = bytesOut + byteBuffer.limit();

		getValueResponseOut++;

		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);
		datagramSocket.send(p);
	}

	public void sendGetValueResponseNodes(InetAddress remoteAddress, int remotePort, int messageId, Identifier valueId, DHTNodeRow node)
			throws IOException
	{
		ByteBuffer byteBuffer = encodeGetValueResponseNodes(messageId, valueId, node);

		bytesOut = bytesOut + byteBuffer.limit();

		getValueResponseNodesOut++;
		DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.remaining(), remoteAddress, remotePort);
		datagramSocket.send(p);
	}

	public void start()
	{
		scheduledExecutor.execute(new Receive());
	}

	public void stopReceiving()
	{
		receiving = false;
		try
		{
			sendPingRequest(getLocalAddress(), getLocalPort());
		}
		catch (IOException e1)
		{
			throw new RuntimeException(e1);
		}
	}

	public void startStatisticGathering(long period)
	{
		scheduledExecutor.scheduleWithFixedDelay(new StatisticsGathering(), 0, period, TimeUnit.SECONDS);
	}

	public LinkedList<int[]> getStatistics()
	{
		return new LinkedList<int[]>(statistics);
	}

	public Map<Identifier, Integer> getFindValueStatistics()
	{
		return new HashMap<Identifier, Integer>(findValueStatistics);
	}

	private void receive() throws IOException
	{

		while (receiving)
		{
			ByteBuffer byteBuffer = ByteBuffer.allocate(2048);
			DatagramPacket p = new DatagramPacket(byteBuffer.array(), byteBuffer.capacity());

			datagramSocket.receive(p);

			switch (byteBuffer.get())
			{
				case 0:			 //ping запрос
				{
					pingRequestIn++;
					decodePingRequest(byteBuffer, p);
					break;
				}
				case 1:			 //ping ответ
				{
					pingResponseIn++;
					decodePingResponse(byteBuffer, p);
					break;
				}
				case 2:			 //get nodes запрос
				{
					getNodesRequestIn++;
					decodeGetNodesRequest(byteBuffer, p);
					break;
				}
				case 3:			 //get nodes ответ
				{
					getNodesResponseIn++;
					decodeGetNodesResponse(byteBuffer, p);
					break;
				}
				case 4:			 //put value запрос
				{
					putValueRequestIn++;
					decodePutValueRequest(byteBuffer, p);
					break;
				}
				case 5:			 //put value ответ
				{
					putValueResponseIn++;
					decodePutValueResponse(byteBuffer, p);
					break;
				}
				case 6:			 //get value запрос
				{
					getValueRequestIn++;
					decodeGetValueRequest(byteBuffer, p);
					break;
				}
				case 7:			 //get value ответ
				{
					getValueResponseIn++;
					decodeGetValueResponse(byteBuffer, p);
					break;
				}
				case 8:			 //get value nodes ответ
				{
					getValueResponseNodesIn++;
					decodeGetValueResponseNodes(byteBuffer, p);
					break;
				}
				default:
					throw new AssertionError("Неизвестный тип сообщения");
			}

		}
	}

	private ByteBuffer encodePingRequest(int messageId)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);
		byteBuffer.put(MessageType.PING_REQUEST.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);	//id запроса
		byteBuffer.flip();
		return byteBuffer;
	}

	private void decodePingRequest(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		bytesIn = bytesIn + byteBuffer.position();
		firePingRequest(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray), messageId);


	}

	private ByteBuffer encodePingResponse(int messageId)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);

		byteBuffer.put(MessageType.PING_RESPONSE.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);						   //id запроса
		byteBuffer.flip();
		return byteBuffer;
	}

	private void decodePingResponse(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		bytesIn = bytesIn + byteBuffer.position();
		outboundMessages.remove(messageId);
		firePingResponse(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray));

	}

	private ByteBuffer encodeGetNodesRequest(int messageId, Identifier searchingId)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);

		byteBuffer.put(MessageType.GET_NODES_REQUEST.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);						   //id запроса
		byteBuffer.put(searchingId.getByteArray());			 //искомый идентификатор
		byteBuffer.flip();
		return byteBuffer;
	}

	private void decodeGetNodesRequest(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		byte[] data = new byte[idLen];
		byteBuffer.get(data);
		bytesIn = bytesIn + byteBuffer.position();
		Identifier searchingIdentifier = Identifiers.getIdentifierFromByteArray(data);
		fireGetNodesRequest(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray), messageId, searchingIdentifier);

	}

	private ByteBuffer encodeGetNodesResponse(int messageId, List<DHTNodeRow> nodes)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);

		byteBuffer.put(MessageType.GET_NODES_RESPONSE.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);						   //id запроса
		byteBuffer.put((byte) nodes.size());						//количество найденных узлов
		for (DHTNodeRow node : nodes)
		{
			byteBuffer.put(node.getInetAddress().getAddress()); //адрес найденного узла
			byteBuffer.putInt(node.getPort());				  //порт найденного узла
		}
		byteBuffer.flip();

		return byteBuffer;
	}

	private void decodeGetNodesResponse(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		outboundMessages.remove(messageId);
		byte size = byteBuffer.get();
		byte[] tempByteArray = new byte[4];
		LinkedList<DHTNodeContacts> list = new LinkedList<DHTNodeContacts>();
		InetAddress ia;
		int port;
		for (int i = 0; i < size; i++)
		{
			byteBuffer.get(tempByteArray);
			ia = InetAddress.getByAddress(tempByteArray);
			port = byteBuffer.getInt();
			list.add(new DHTNodeContacts(ia, port));
		}
		bytesIn = bytesIn + byteBuffer.position();
		fireGetNodesResponse(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray), list);

	}

	private ByteBuffer encodePutValueRequest(int messageId, Identifier valueId, String value)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);

		byteBuffer.put(MessageType.PUT_VALUE_REQUEST.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);						   //id запроса
		byteBuffer.put(valueId.getByteArray());				 //идентификатор значения
		byteBuffer.put((byte) value.getBytes().length);			 //длина массива байтов со значением
		byteBuffer.put(value.getBytes());					   //массив байтов со значением
		byteBuffer.flip();

		return byteBuffer;

	}

	private void decodePutValueRequest(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		byte[] data = new byte[idLen];
		byteBuffer.get(data);
		Identifier valueIdentifier = Identifiers.getIdentifierFromByteArray(data);
		byte[] valueArray = new byte[byteBuffer.get()];
		byteBuffer.get(valueArray);
		bytesIn = bytesIn + byteBuffer.position();
		String value = new String(valueArray);
		firePutValueRequest(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray), messageId, valueIdentifier, value);

	}

	private ByteBuffer encodePutValueResponse(int messageId)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);

		byteBuffer.put(MessageType.PUT_VALUE_RESPONSE.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);						   //id запроса
		byteBuffer.flip();
		return byteBuffer;
	}

	private void decodePutValueResponse(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		bytesIn = bytesIn + byteBuffer.position();
		outboundMessages.remove(messageId);
		firePutValueResponse(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray));

	}

	private ByteBuffer encodeGetValueRequest(int messageId, Identifier valueId)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);

		byteBuffer.put(MessageType.GET_VALUE_REQUEST.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);						   //id запроса
		byteBuffer.put(valueId.getByteArray());				 //идентификатор значения
		byteBuffer.flip();

		return byteBuffer;
	}

	private void decodeGetValueRequest(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		byte[] data = new byte[idLen];
		byteBuffer.get(data);
		bytesIn = bytesIn + byteBuffer.position();
		Identifier valueIdentifier = Identifiers.getIdentifierFromByteArray(data);
		fireGetValueRequest(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray), messageId, valueIdentifier);

	}

	private ByteBuffer encodeGetValueResponse(int messageId, List<DHTDataRow> values)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);

		byteBuffer.put(MessageType.GET_VALUE_RESPONSE.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);						   //id запроса
		byteBuffer.put((byte) values.size());						//количество найденных значений
		for (DHTDataRow value : values)
		{
			byteBuffer.put((byte) value.getValue().getBytes().length);	  //размер найденного значения
			byteBuffer.put(value.getValue().getBytes());				   //найденное значение
			byteBuffer.putLong(value.getTtl());				 //TTL найденного значения
		}
		byteBuffer.flip();

		return byteBuffer;
	}

	private void decodeGetValueResponse(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		outboundMessages.remove(messageId);
		List<DHTDataRow> values = new LinkedList<DHTDataRow>();
		byte size = byteBuffer.get();
		for (byte i = 0; i < size; i++)
		{
			byte valueSize = byteBuffer.get();
			byte[] value = new byte[valueSize];
			byteBuffer.get(value);
			long ttl = byteBuffer.getLong();
			values.add(new DHTDataRow(Identifiers.createIdentifier(new String(value), idLen), new String(value), ttl));
		}
		bytesIn = bytesIn + byteBuffer.position();
		fireGetValueResponse(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray), values);

	}

	private ByteBuffer encodeGetValueResponseNodes(int messageId, Identifier valueId, DHTNodeRow node)
	{
		ByteBuffer byteBuffer = ByteBuffer.allocate(2048);

		byteBuffer.put(MessageType.GET_VALUE_RESPONSE_NODES.getNumber());								//тип сообщения
		byteBuffer.put(localId.getByteArray());				 //id отправителя
		byteBuffer.putInt(messageId);						   //id запроса
		byteBuffer.put(valueId.getByteArray());				 //id ненайденного значения
		byteBuffer.put(node.getInetAddress().getAddress()); //адрес найденного узла
		byteBuffer.putInt(node.getPort());				  //порт найденного узла
		byteBuffer.flip();
		return byteBuffer;
	}

	private void decodeGetValueResponseNodes(ByteBuffer byteBuffer, DatagramPacket p) throws IOException
	{
		byte[] identifierArray = new byte[idLen];
		int messageId;
		byteBuffer.get(identifierArray);
		messageId = byteBuffer.getInt();
		outboundMessages.remove(messageId);
		byte[] valueArray = new byte[idLen];
		byteBuffer.get(valueArray);
		Identifier valueId = Identifiers.getIdentifierFromByteArray(valueArray);
		byte[] tempByteArray = new byte[4];
		InetAddress ia;
		int port;
		byteBuffer.get(tempByteArray);
		ia = InetAddress.getByAddress(tempByteArray);
		port = byteBuffer.getInt();
		bytesIn = bytesIn + byteBuffer.position();
		fireGetValueResponseNodes(p.getAddress(), p.getPort(), Identifiers.getIdentifierFromByteArray(identifierArray), valueId, new DHTNodeContacts(ia, port));

	}


	private void firePingRequest(InetAddress address, int port, Identifier id, int messageId) throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receivePingRequest(address, port, id, messageId);
		}
	}

	private void firePingResponse(InetAddress address, int port, Identifier id) throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receivePingResponse(address, port, id);
		}
	}

	private void fireGetNodesRequest(InetAddress address, int port, Identifier id, int messageId, Identifier searchingIdentifier)
			throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receiveGetNodesRequest(address, port, id, messageId, searchingIdentifier);
		}
	}

	private void fireGetNodesResponse(InetAddress address, int port, Identifier id, List<DHTNodeContacts> list)
			throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receiveGetNodesResponse(address, port, id, list);
		}
	}

	private void firePutValueRequest(InetAddress address, int port, Identifier id, int messageId, Identifier valueId, String value)
			throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receivePutValueRequest(address, port, id, messageId, valueId, value);
		}
	}

	private void firePutValueResponse(InetAddress address, int port, Identifier id) throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receivePutValueResponse(address, port, id);
		}
	}

	private void fireGetValueRequest(InetAddress address, int port, Identifier id, int messageId, Identifier valueId)
			throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receiveGetValueRequest(address, port, id, messageId, valueId);
		}
	}

	private void fireGetValueResponse(InetAddress address, int port, Identifier id, List<DHTDataRow> values)
			throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receiveGetValueResponse(address, port, id, values);
		}
	}

	private void fireGetValueResponseNodes(InetAddress address, int port, Identifier id, Identifier valueId, DHTNodeContacts nodes)
			throws IOException
	{
		for (NetworkListener listener : listeners)
		{
			listener.receiveGetValueResponseNodes(address, port, id, valueId, nodes);
		}
	}

	private void fireReceiveTimeout(Integer messageId)
	{
		DHTNodeContacts timeoutNode = outboundMessages.get(messageId);
		System.out.println("timeout");
		if (timeoutNode != null)
		{
			for (NetworkListener listener : listeners)
			{
				listener.receiveTimeout(timeoutNode);
			}
		}
	}

	private class TimeoutTask implements Runnable
	{

		private final int messageNumber;

		TimeoutTask(int messageNumber)
		{
			this.messageNumber = messageNumber;
		}


		public void run()
		{
			fireReceiveTimeout(messageNumber);
		}
	}

	private class Receive implements Runnable
	{

		public void run()
		{
			try
			{
				receive();
			}
			catch (IOException e)
			{
				throw new RuntimeException(e);
			}
		}
	}

	private final class StatisticsGathering implements Runnable
	{

		//@Override
		public void run()
		{
			statistics.add(new int[]{
					pingRequestIn, pingResponseIn,
					getNodesRequestIn, getNodesResponseIn,
					putValueRequestIn, putValueResponseIn,
					getValueRequestIn, getValueResponseIn, getValueResponseNodesIn,
					pingRequestOut, pingResponseOut,
					getNodesRequestOut, getNodesResponseOut,
					putValueRequestOut, putValueResponseOut,
					getValueRequestOut, getValueResponseOut, getValueResponseNodesOut,
					bytesIn, bytesOut
			});
		}

	}

	private enum MessageType
	{

		PING_REQUEST((byte) 0), PING_RESPONSE((byte) 1),
		GET_NODES_REQUEST((byte) 2), GET_NODES_RESPONSE((byte) 3),
		PUT_VALUE_REQUEST((byte) 4), PUT_VALUE_RESPONSE((byte) 5),
		GET_VALUE_REQUEST((byte) 6), GET_VALUE_RESPONSE((byte) 7), GET_VALUE_RESPONSE_NODES((byte) 8);

		private final byte number;

		MessageType(byte number)
		{
			this.number = number;
		}

		private byte getNumber()
		{
			return number;
		}
	}
}


